package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2023/4/10
 * 每隔5min输出最近1h内点击量最多的前3个商品。
 *
 * 输入: 一个用户对一个商品的一次pv浏览
 *
 *    计算过程:
 *             滑动窗口。
 *          ①第一次聚合
 *              分组: 商品id
 *              开窗，汇总统计。统计最近1h，所有用户，对每个商品的 总点击次数
 *                  输入:  11--a--[8:20]--pv
 *                        12--a--[8:21]--pv
 *
 *                  输出: [8,9],a-2
 *
 *              计算的时机: 到窗口的endTime点就算。
 *
 * 输出： 点击量最多的前3个商品
 *         ②第二次聚合，根据同一个窗口中，所有商品的点击总次数，排序，求前3
 *              分组: 按照窗口的范围
 *          [8,9]-a-100
 *          [8,9]-b-101
 *          [8,9]-c-102
 *          [8,9]-d-103
 *
 *          [9,10]-a-10
 *          [9,10]-b-12
 *                  不是窗口运算，就是普通的聚合。
 *                  普通的聚合，是来一条数据就聚合一次！为了保证效果，必须等当前窗口范围的数据全部到达，再聚合，
 *                  只有通过定时器来实现。
 *
 */
public class Demo11_TopN
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        FileSource<String> source = FileSource.forRecordStreamFormat(
            new TextLineInputFormat(),
            new Path("data/UserBehavior.csv")
        ).build();

        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy
            .<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTimestamp());


        //1.读数据，生成水印
        SingleOutputStreamOperator<UserBehavior> ds = env
            .fromSource(source, WatermarkStrategy.noWatermarks(), "source")
            .map(new MapFunction<String, UserBehavior>()
            {
                @Override
                public UserBehavior map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new UserBehavior(
                        Long.valueOf(words[0]),
                        Long.valueOf(words[1]),
                        Integer.valueOf(words[2]),
                        words[3],
                        Long.valueOf(words[4]) * 1000
                    );
                }
            })
            //过滤页面点击浏览
            .filter(u -> "pv".equals(u.getBehavior()))
            .assignTimestampsAndWatermarks(watermarkStrategy);
        
        /*
            2.第一次聚合，按照 商品id分组，开窗运算

            输入: UserBehavior
            输出:  [8,9],a,30
                不是同一种类型，还希望使用 滚动聚合，没得选，只有aggregate
         */
        SingleOutputStreamOperator<HotItem> ds1 = ds
            .keyBy(UserBehavior::getItemId)
            .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
            .aggregate(new AggregateFunction<UserBehavior, Integer, HotItem>()
            {
                @Override
                public Integer createAccumulator() {
                    return 0;
                }

                @Override
                public Integer add(UserBehavior value, Integer accumulator) {
                    return accumulator + 1;
                }

                @Override
                public HotItem getResult(Integer accumulator) {
                    return new HotItem(null, null, null, accumulator);
                }

                @Override
                public Integer merge(Integer a, Integer b) {
                    return null;
                }
            }, new WindowFunction<HotItem, HotItem, Long, TimeWindow>()
            {
                /*
                        输入就是上述AggregateFunction的最终输出
                 */
                @Override
                public void apply(Long itemId, TimeWindow window, Iterable<HotItem> input, Collector<HotItem> out) throws Exception {
                    HotItem hotItem = input.iterator().next();
                    hotItem.setStart(window.getStart());
                    hotItem.setEnd(window.getEnd());
                    hotItem.setItemId(itemId);
                    out.collect(hotItem);

                }
            });

        //3.第二次聚合  按照窗口范围分组，再进行普通聚合。使用定时器，指定一个在窗口的end时间触发的定时器，进行top3的排序
        ds1
            .keyBy(HotItem::getEnd)
            .process(new KeyedProcessFunction<Long, HotItem, String>()
            {

                private ValueState<Boolean> ifFirst;
                private ListState<HotItem> hotitems;

                @Override
                public void open(Configuration parameters) throws Exception {
                    hotitems = getRuntimeContext().getListState(new ListStateDescriptor<>("hotitems", HotItem.class));
                    ifFirst = getRuntimeContext().getState(new ValueStateDescriptor<>("ifFirst", Boolean.class));
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                    //对已经存储在状态中的数据，求top3
                    List<HotItem> top3HotItem = StreamSupport.stream(hotitems.get().spliterator(), true)
                                                         .sorted((h1, h2) -> -h1.getCount().compareTo(h2.getCount()))
                                                         .limit(3)
                                                         .collect(Collectors.toList());

                    String top3Str = top3HotItem.stream().map(h -> h.getItemId() + ":" + h.getCount()).collect(Collectors.joining(","));
                    TimeWindow w = null;
                    if (top3HotItem.size() > 0){
                        w = new TimeWindow(top3HotItem.get(0).getStart(),top3HotItem.get(0).getEnd());
                    }

                    if (w != null){
                        out.collect(MyUtil.parseTimeWindow(w) + " top3 : "+top3Str);

                    }

                }

                /*
                                        来一条数据就调用一次

                                        使用定时器，就算top3 ： 一个窗口，只计算一次。
                                                                    弊端，消耗状态的存储。

                                        使用Demo5_KeyedListState 计算top3的区别: 来一条数据，就计算一次。向下游输出的数据多。
                                                            可以没来一个，就保持状态中，只存3个，节省状态的存储空间。
                                 */
                @Override
                public void processElement(HotItem value, Context ctx, Collector<String> out) throws Exception {

                    TimerService timerService = ctx.timerService();
                    //加入集合中先缓存着
                    hotitems.add(value);

                    //如果是当前窗口时间段的一条数据，就指定定时器
                    if (ifFirst.value() == null){
                        timerService.registerEventTimeTimer(value.getEnd());
                        ifFirst.update(false);
                    }
                }
            })
            .print();
        


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class HotItem{
        //声明窗口范围
        private Long start;
        private Long end;
        private Long itemId;
        private Integer count;
    }
}
